package com.xz.bigsoubackend;


import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.xz.bigsoubackend.esdao.PostEsDao;
import com.xz.bigsoubackend.model.dto.post.PostEsDTO;
import com.xz.bigsoubackend.service.PostService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author xuzhou
 * @version 1.0
 * @createTime 2024-11-22
 * @description
 */

@Service
@Slf4j
public class CanalReceiver {
    @Autowired
    @Qualifier("canalConnector")
    private CanalConnector canalConnector;
    @Value("${canal.subscribe}")
    private String subscribe;

    @Resource
    private PostEsDao postEsDao;

    @PostConstruct
    public void connect() {
        canalConnector.connect();
        canalConnector.subscribe(subscribe);
        canalConnector.rollback();
    }

    @PreDestroy
    public void disConnect() {
        canalConnector.disconnect();
    }

    @Async
    @Scheduled(initialDelayString = "${canal.initialDelay:3000}", fixedDelayString = "${canal.fixedDelay:3000}")
    public void processData() {
        try {
            if (!canalConnector.checkValid()) {
                log.warn("=====>与Canal服务器的连接失效！！！重连，下个周期再检查数据变更");
                this.connect();
            } else {
                //获取batchSize条数据
                Message message = canalConnector.getWithoutAck(100);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    log.info("=====>本批次[{}]没有数据需要同步", batchId);
                } else {
                    log.info("=====>本批次[{}]数据同步共有[{}]个更新需要处理", batchId, size);
                    for (CanalEntry.Entry entry : message.getEntries()) {
                        log.info(entry.getEntryType().toString());
                        if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                                || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                            log.info("=====>当前语句为事务开始或者事务结束, 不做处理");
                            continue;
                        }
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                        //获取表名
                        String tableName = entry.getHeader().getTableName();
                        CanalEntry.EventType eventType = rowChange.getEventType();
                        log.info("=====>数据变更详情：来自binglog[{}.{}], 数据源{}.{}, 变更类型{}",
                                entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                                entry.getHeader().getSchemaName(), tableName, eventType);

                        //处理数据
                        handleData(entry);
                    }

                    // 提交确认
                    canalConnector.ack(batchId);
                    log.info("=====>本批次[{}]Canal同步数据完成", batchId);
                }
            }
        } catch (Exception e) {
            log.error("=====>Canal同步数据失效，请检查：", e);
        }
    }


    private void handleData(CanalEntry.Entry entry ) {
        //自己的逻辑，比如保存到数据库，推送的mq，保存到redis
        System.out.println(entry);
        CanalEntry.RowChange rowChage = null;
        try {
            rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (Exception e) {
            throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                    e);
        }
        String tableName = entry.getHeader().getTableName();
        CanalEntry.EventType eventType = rowChage.getEventType();
        List<CanalEntry.RowData> rowDataList = rowChage.getRowDatasList();
        String schemaName = entry.getHeader().getSchemaName();
        if (!("my_db".equals(schemaName) && "post".equals(tableName))){
            return;
        }
        // 处理数据变更事件
        for (CanalEntry.RowData rowData : rowDataList) {
            switch (eventType) {
                case INSERT:
                    // 处理插入事件
                    dealInsert(schemaName, tableName, rowData.getAfterColumnsList());
                    break;
                case UPDATE:
                    // 处理更新事件
                    dealUpdate(schemaName, tableName, rowData.getAfterColumnsList());
                    break;
                case DELETE:
                    // 处理删除事件
                    dealDelete(schemaName, tableName, rowData.getBeforeColumnsList());
                    break;
                default:
                    break;
            }
        }
        log.info("=====>Canal同步数据处理完成");
    }

    private SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private void dealDelete(String schemaName, String tableName, List<CanalEntry.Column> afterColumnsList) {
        Map<String, Object> dataMap = new HashMap<>();
        for (CanalEntry.Column column : afterColumnsList) {
            dataMap.put(column.getName(), column.getValue());
        }
//        log.debug("delate data:{}", afterColumnsList);
        log.debug("delate map data:{}", dataMap);
    }

    private void dealUpdate(String schemaName, String tableName, List<CanalEntry.Column> columns) {
        try {
            PostEsDTO postEsDTO = new PostEsDTO();
            postEsDTO.setId(Long.parseLong(columns.get(0).getValue()));
            postEsDTO.setTitle(columns.get(1).getValue());
            postEsDTO.setContent(columns.get(2).getValue());

            // Arrays.asList(.split(","))
            postEsDTO.setTags(JSONUtil.toList(columns.get(3).getValue(), String.class));
            postEsDTO.setUserId(Long.parseLong(columns.get(6).getValue()));
            String value = columns.get(8).getValue();
            postEsDTO.setUpdateTime(formatter.parse(value));
            postEsDTO.setCreateTime(formatter.parse(columns.get(7).getValue()));
            postEsDTO.setIsDelete(Integer.valueOf(columns.get(9).getValue()));
            postEsDao.save(postEsDTO);
            System.out.println(postEsDTO.getId());
        } catch (ParseException e) {
            log.error(e.getMessage(), "canal 将 数据写入 es 失败");
            throw new RuntimeException(e);
        }
//        log.debug("update data:{}", columns);
    }

    private void dealInsert(String schemaName, String tableName, List<CanalEntry.Column> columns) {
        try {
            PostEsDTO postEsDTO = new PostEsDTO();
            postEsDTO.setId(Long.parseLong(columns.get(0).getValue()));
            postEsDTO.setTitle(columns.get(1).getValue());
            postEsDTO.setContent(columns.get(2).getValue());

            // Arrays.asList(.split(","))
            postEsDTO.setTags(JSONUtil.toList(columns.get(3).getValue(), String.class));
            postEsDTO.setUserId(Long.parseLong(columns.get(6).getValue()));
            String value = columns.get(8).getValue();
            postEsDTO.setUpdateTime(formatter.parse(value));
            postEsDTO.setCreateTime(formatter.parse(columns.get(7).getValue()));
            postEsDTO.setIsDelete(Integer.valueOf(columns.get(9).getValue()));
            postEsDao.save(postEsDTO);
            System.out.println(postEsDTO.getId());
        } catch (ParseException e) {
            log.error(e.getMessage(), "canal 将 数据写入 es 失败");
            throw new RuntimeException(e);
        }
    }
}